import java.util
import org.apache.spark.mllib.recommendation.{ALS, Rating}
import org.apache.spark.{SparkConf, SparkContext}
object My15 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("UserProRecom")//.setMaster("local")
    val sc = new SparkContext(conf)
    //1.加载数据
    val data = sc.textFile(args(0))//"F://浪潮//课设//data//DxFileProHigh"
    val userlid = sc.textFile(args(1)) //"F://浪潮//课设//data//DxFileUExtract"         
    //2.userid, userid hash 值 | 预购类型:产品访问次数,预购类型:产品访问次数
    val userId  = userlid.map(x => {
      val arr = x.split("\\|")
      (arr(0),arr(0).hashCode+"|"+arr(1))   //用户ID,用户ID hash 值 | 预购类型:产品访问次数,预购类型:产品访问次数
    }).cache()
    //userId.unpersist()
    //3.输出数据：userFlat <string>: {预购类型，用户id,用户id hash值}
    val userFlat = userId.flatMap( x => {           
      val arr = x._2.split("\\|")
      val arr2 = arr(1).split(",")
      val outList = ListBuffer[String]()
      val arr3 = arr2(0).split(":")//1
      if(arr2.length>1){
         val arr4 = arr2(1).split(":")//2
         outList+=(arr3(0)+","+x._1+","+arr(0))
         outList+=(arr4(0)+","+x._1+","+arr(0))
      }
      else{
        outList+=(arr3(0)+","+x._1+","+arr(0))
      }
      outList
    })
    //4.用正则替换掉产品id里的非数字字符
    val basicId = data.map(x=>{                             
      val arr = x.split("\\|")
      val re = arr(0).replaceAll("[^0-9]","")
      (arr(1),re+","+arr(2)+","+arr(3))//用户ID；产品ID(int 值),访问次数,预购类型
    }).cache()
    // basicId.unpersist()
    //5.关联第4步 basicId 与第2步 userId 
    val basicdata = basicId.join(userId).map(x => {
      (x._1,(x._2._1,x._2._2.split("\\|")(0)+","+x._2._2.split("\\|")(1)))
      //用户ID；(产品ID，访问次数,预购类型; 用户id int型hash值，
      //预购类型:产品访问次数,预购类型:产品访问次数)
    })
    //6.按spark协同过滤数据格式转换数据，获取Rating
   val ratings = basicdata.map(x => {
      new Rating(x._2._2.split(",")(0).toInt, x._2._1.split(",")(0).toInt, x._2._1.split(",")(1).toDouble)
      //用户ID的int hash值、产品ID、打分(访问次数)
    })
    //7.生成推荐模型
    val rank = 5						//设置隐藏因子
    val numIterations = 10					//设置迭代次数
    val model = ALS.train(ratings, rank, numIterations, 0.01)	//进行模型训练
    //8.加载所有产品
    val promsg = sc.textFile(args(2))//"F://浪潮//课设//configData//car_type.txt"
    //9.生成产品字典数据的元组
    val proPair = promsg.map(x =>{
      val arr = x.split("\\|")
     (arr(2),arr(1)) //产品类型; 产品id
    })
    //10.转换用户预购数据的表达形式
    val userProType = userFlat.map(x=>{
      val arr = x.split(",")
      (arr(0),arr(1)+","+arr(2))        //预购类型；用户id,用户intID
    })
    //11.用户与产品的全集（统计用户对应产品分类有数据的产品）
    val userProScore = userProType.join(proPair).map(x=>{
      (x._2._1.split(",")(1).toInt,x._2._2.replaceAll("[^0-9]","").toInt)  //userId hash值；产品id int值
    }).distinct
    //12.预测用户对所有产品的评分
    val userProPair  = model.predict(userProScore).map(x => (x.user , x.product+","+x.rating))
    //13.获取userid hash 与 userid的键值对
    val userInt  = userId.map(x=>{
      (x._2.split("\\|")(0).toInt,x._1)//userid hash; userid
    })
    //14.生成 产品对应的用户id、评分
    val userJoin = userProPair.join(userInt).map(x=>{
      (x._1,(x._2._1.split(",")(0)+","+x._2._1.split(",")(1),x._2._2))  
      //{userid hash值; { int型产品id,评分;userid }}
    })   
    val userScore = userJoin.map(x=>{
      (x._2._1.split(",")(0).toInt,x._2._2+","+x._2._1.split(",")(1))
      //int型产品id ; 用户id，评分
    })   
    //15.将第8步的所有产品字典数据字符串形式转为产品id int值对应的产品信息。
    val proClass = promsg.map(x=>{
      val arr = x.split("\\|")
      (arr(1).replaceAll("[^0-9]","").toInt,arr(0)+","+arr(1))//int型产品id;产品分类，产品id
    })
    //16.生成待排行的数据键值对
    val userProMsg = userScore.join(proClass).map(x=>{
      (x._2._1.split(",")(0)+","+x._2._2.split(",")(0),x._2._2.split(",")(1)+","+x._2._1.split(",")(1))  
      //用户id，产品分类；产品id，评分
    })
    //17.按用户、预购类型分类，每类别取评分前5
    val userTypeTop  = userProMsg.groupByKey.map(x => {
      val ite = x._2.iterator
      val topMap : util.TreeMap[Double, String] = new util.TreeMap[Double, String]
      while (ite.hasNext) {
        val proPirce = ite.next.split(",")
        if (proPirce(1).toDouble > 0) {
          topMap.put(proPirce(1).toDouble, proPirce(0))
        }
        if (topMap.size > 5) {
          topMap.remove(topMap.firstKey)
        }
      }

      val userTypeTop2 = ListBuffer[String]()
      val sb = new StringBuffer
      import scala.collection.JavaConversions._
      for (key <- topMap.keySet) {
        sb.append(topMap.get(key)).append(",")
      }
      if(sb!=null && sb.length()>0){
        x._1 + "," + sb.toString.substring(0, sb.toString.length - 1)
      }    
    }).distinct
    userTypeTop.saveAsTextFile(args(3))//"F://浪潮//课设//data//15.DxALS2"
    sc.stop()
    
    }
}